package com.idroc.message.filter.codec;

import java.util.LinkedList;
import java.util.List;

import org.apache.mina.common.ByteBuffer;
import org.apache.mina.common.IoSession;
import org.apache.mina.filter.codec.ProtocolDecoder;
import org.apache.mina.filter.codec.ProtocolDecoderOutput;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.idroc.message.Message;
import com.idroc.message.MessageType;

/**
 * 多路复用解码器
 * @author 赤脚詹(yuyoo4j@163.com)
 * @date Aug 17, 2009 4:13:38 PM 
 * @memo:
 */
@SuppressWarnings("unchecked")
public class MessageDecoder implements ProtocolDecoder {
	
	private static final Logger log = LoggerFactory.getLogger(MessageDecoder.class);
	/**
	 * 消息上下文保存在mina IoSession属性中的key
	 */
	private static final String CHANNEL_INPUT_CONTEXT = "channel_input_context"; 
	/**
	 * 消息通讯版本
	 */
	private static final byte MSG_COMMUNICATION_VERSION = (byte)1;

	/**
	 * 解码
	 */
	public void decode(IoSession session, ByteBuffer bytebuffer,
			ProtocolDecoderOutput output) throws Exception {
		
		log.debug("decode......");
		
		Context context = getContext(session, bytebuffer);

		log.debug("Context. context.newArrives.size:{}", context.newArrives.size()); 
		log.debug("Context. context.state:{}", context.state);
		log.debug("Context. context.totalLength:{}", context.totalLength);		
		log.debug("Context. context.readLength:{}", context.readLength);	
		log.debug("Context. context.msgBuffer.position:{}, context.msgBuffer:{}", context.msgBuffer.position(), context.msgBuffer);
				
		for (Message msg : context.newArrives) {
			try {
				output.write(msg);
			} catch (Exception ex) { 
				context.errorMessages.add(msg);
				log.error("decode Exception", ex);
			} 
		}		 
		context.newArrives.clear();
		if (context.errorMessages.size() > 0) {
			String exStr = context.errorMessages.toString();
			context.errorMessages.clear();
			throw new RuntimeException(exStr);
		}
	}

	public void dispose(IoSession iosession) throws Exception {

	}

	public void finishDecode(IoSession iosession,
			ProtocolDecoderOutput protocoldecoderoutput) throws Exception {

	}
	
	/**
	 * 获取解码上下文
	 * @param session -- mina IoSession
	 * @param buffer -- 当前输入 mina ByteBuffer
	 * @return -- 上下文
	 */
	private Context getContext(IoSession session, ByteBuffer buffer) {
		
		Context context= (Context)session.getAttribute(CHANNEL_INPUT_CONTEXT);		
		if (null == context) {
			context = new Context();
			session.setAttribute(CHANNEL_INPUT_CONTEXT, context);
		}
				
		do {			 
			if (context.state == Context.DataState.COMPLETE) {  // 消息完全到达
				Context.doComplete(context, buffer);
			} else if (context.state == Context.DataState.SEGMENT) { // 消息只到达部分
				Context.doSegment(context, buffer);
			}
		} while (buffer.hasRemaining());
				
		return context;
	}
	
	/**
	 * 解码上下文
	 * @author: tianya.cn 
	 * @creator: zhandl (zhandl@hainan.net)
	 * @date: May 15, 2009 4:41:28 PM
	 * @memo: 包括命令信息, 总长度, 当前长度, 有效数据ByteBuffer
	 */
	private static class Context {
		
		/**
		 * 数据状态
		 * @author: tianya.cn 
		 * @creator: zhandl (zhandl@hainan.net)
		 * @date: Jun 5, 2009 1:44:35 PM
		 * @memo: SEGMENT: 数据片段状态,已读取的数据存在残缺;COMPLETE: 数据块完整状态, 已读取的消息处于完整状态.
		 */
		private enum DataState{ SEGMENT, COMPLETE}		
		/**
		 * 当前消息状态
		 */
		DataState state = DataState.COMPLETE;		
		/**
		 * 是否包含完整消息头
		 */
		boolean isFullMessageHead = true;
		/**
		 * 消息头缓存数组
		 */
		byte[] messageHeadCache = new byte[5];
		/**
		 * 已经读取的消息头字节数
		 */
		int readMessageHeadBytes = 0; 
		/**
		 * 消息总长度
		 */
		int totalLength = 0; 
		/**
		 * 已经读取的消息长度
		 */
		int readLength = 0;
		/**
		 * 消息包内容
		 */
		ByteBuffer msgBuffer = ByteBuffer.allocate(2048).setAutoExpand(true);
		/**
		 * 新到达消息列表
		 */
		List<Message> newArrives = new LinkedList<Message>();
		/**
		 * 错误数据列表
		 */
		List<Message> errorMessages = new LinkedList<Message>();
		
		/**
		 * 处理处于数据残缺状态的数据
		 * @param context -- 消息上下文
		 * @param ib -- 消息输入流缓存
		 * @return -- 消息上下文
		 */
		static Context doSegment(Context context, ByteBuffer ib) {
			
			int remaining = ib.remaining(); 
			
			int remain = context.totalLength - context.readLength;
			if (remain <= remaining) {
				
				// 获取消息剩余的数据片段
				ByteBuffer msgSegment = ib.slice();
				msgSegment.limit(remain);
				ib.position(ib.position() + remain);
				context.msgBuffer.put(msgSegment);
				
				context.readLength += remain;
				if (context.readLength != context.msgBuffer.position()) {
					log.error("context.readLength:{}, context.msgBuffer.position:{}", context.readLength, context.msgBuffer.position());
				}
								
				// 解码buffer的数据
				context.msgBuffer.flip();
				Message msg = decodeMsg(context.msgBuffer);
				if (null != msg) {
					context.newArrives.add(msg);
				}
							
				// 设置上下文回初始状态
				context.totalLength = 0;
				context.readLength = 0;
				context.msgBuffer.rewind();
				context.state = DataState.COMPLETE;
			} else {
				context.readLength += ib.remaining();
				context.msgBuffer.put(ib); 
			}
			return context;
		}
		
		/**
		 * 处理处于完整状态的消息输入流缓存
		 * @param context -- 消息上下文
		 * @param ib -- 消息输入流缓存
		 * @return -- 消息上下文
		 */
		static Context doComplete(Context context, ByteBuffer ib) {
			 
			// 如果已经成功读取消息长度
			if (readMessageHead(context, ib)) {		
							
				if (context.totalLength <= ib.remaining()) {
					
					// 在原有的buffer上构造新的buffer,此buffer用于包含已经完整的消息数据
					ByteBuffer msgBuffer = ib.slice();
					ib.position(ib.position() + context.totalLength);
					msgBuffer.limit(context.totalLength);
										
					// 解码buffer的数据
					Message msg = decodeMsg(msgBuffer);
					if (null != msg) {
						context.newArrives.add(msg);
					}
					// 设置上下文回初始状态
					context.totalLength = 0;
					context.readLength = 0;
					context.msgBuffer.rewind();
					context.state = DataState.COMPLETE;
				} else { 
					// 保存消息片段数据
					context.readLength = ib.remaining();
					context.msgBuffer.put(ib);
					context.state = DataState.SEGMENT;
				}				
			}
			return context;
		}
		
		/**
		 * 对消息解码
		 * @param msgBuffer -- 包含消息数据的buffer
		 * @return 消息对象
		 */
		private static Message decodeMsg(ByteBuffer msgBuffer) {
			
			// 解码buffer的数据
			msgBuffer.mark();
			int msgId = msgBuffer.getInt();
			msgBuffer.reset();
			
			Message msg = null;
			MessageType msgType = null;
			Class msgMetadata = null;
			try {
				// 获取消息类型,并对消息解码
				msgType = MessageType.get(msgId);
				msgMetadata = msgType.getMsgMeta();
				msg = (Message) msgMetadata.newInstance();
				msg.decode(msgBuffer);
			} catch (Exception ex) {
				 Object[] info = new Object[]{msgType, msgMetadata, msgBuffer};
				 log.error("msgType:{}, msgMetadata:{}, msgBuffer:{}", info);
				 log.error("Exception:", ex);
			}
			return msg;
		} 
		
		/**
		 * 读取消息头
		 * @param context -- 消息上下文
		 * @param ib -- 消息输入字节流
		 * @return true,如果已经读取完整的消息头
		 */
		static boolean readMessageHead(Context context, ByteBuffer ib) {

			boolean rtv = true; 

			int remaining = ib.remaining();
			if (context.isFullMessageHead) {  // 上一次已经读取完整的消息头了. RT
				if (remaining >= 5) {	// 输入流字节缓存允许读取4个字节	4T
					// 检查通信数据版本
					checkCommunicationVersion(ib.get());					
					context.totalLength = ib.getInt();
					
				} else { // 输入流字节缓存没有5个字节  XT
					context.isFullMessageHead = false;
					context.readMessageHeadBytes = remaining;
					ib.get(context.messageHeadCache, 0, remaining);
					rtv = false;
				}
			} else {  // 上一次未全部读取totalLength值了 UT
				if (remaining >= 5) { 	// 输入流字节缓存允许读取4个字节	4T
					for (; context.readMessageHeadBytes < 5; context.readMessageHeadBytes++) {
						context.messageHeadCache[context.readMessageHeadBytes] = ib.get();
					}
					context.readMessageHeadBytes = 0;
					context.totalLength = getInt(context.messageHeadCache, 1);
					// 检查通信数据版本
					checkCommunicationVersion(context.messageHeadCache[0]);	
					context.isFullMessageHead = true; 
					
				} else { // 输入流字节缓存没有4个字节 XT
					ib.get(context.messageHeadCache, context.readMessageHeadBytes, remaining);
					context.readMessageHeadBytes += remaining;
					rtv = false;
				}
			} 
			
			return rtv; 
		}
		
		/**
		 * 将字节数组值转换成Int值
		 * @param intVal -- 字节数组
		 * @return 数据前4个字节的int值
		 */
		private static int getInt(byte[] intVal, int offset) {
			
			int ch1 = intVal[offset + 0];
			int ch2 = intVal[offset + 1];
			int ch3 = intVal[offset + 2];
			int ch4 = intVal[offset + 3];
			
			return ((ch1 << 24) + (ch2 << 16) + (ch3 << 8) + (ch4 << 0));
		}
		
		private static void checkCommunicationVersion(byte msgCommVersion) {
			
			if (msgCommVersion != MSG_COMMUNICATION_VERSION) { // 错误版本号数据	
				
				String msg = "非法数据通信版本号, msgCommVersion:" + msgCommVersion + ", MSG_COMMUNICATION_VERSION:" + MSG_COMMUNICATION_VERSION;
				throw new RuntimeException(msg);
			} 
		}
			
	}
}
